package com.icbc.dccsh.bolt;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.joda.time.DateTime;

import com.icbc.dccsh.util.Watcher;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class FlattenBolt extends BaseBasicBolt  {
	private Watcher _watcher = null;
	private DateFormat _df = null;
//	private DateFormat _df2date = null;
	@Override
    public void prepare(Map stormConf, TopologyContext context) {
		this._watcher = Watcher.getWatcher(this.getClass());
		this._df = new SimpleDateFormat("yyyyMMddHHmmss");
//		this._df2date = new SimpleDateFormat("yyyyMMdd");
    }
	
	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
		HashMap map = (HashMap)input.getValueByField("result");
		this._watcher.count();
		HashMap flattenHash = new HashMap();
		String ip = (String)((HashMap)map.get("endpoint")).get("ip");
		String devicename = (String)((HashMap)map.get("endpoint")).get("devicename");
		String objectname = (String)map.get("object");
		String logDatetime_str = (String)map.get("timestamp");
		String indexName = (String)map.get("indexname");
		long calcDatetimeEpoch = (Long)map.get("calcdatetime");
		
//		String id = (String)map.get("id");
//		flattenHash.put("id", id);
		
		flattenHash.put("ip", ip);
		flattenHash.put("devicename", devicename);
		flattenHash.put("object", objectname);
//		flattenHash.put("calcdatetime", calcDatetimeEpoch);
		Date dt = null;
		Date calcDate = null;
		long timestampEpoch = 0;
		try {
			dt = (Date)this._df.parse(logDatetime_str);
			timestampEpoch = dt.getTime();
			calcDate = new Date(calcDatetimeEpoch);
//			calcDate = this._df.format(new Date(calcdatetime));
		} catch (ParseException e) {
			System.out.println(e.toString());
		}
		flattenHash.put("timestamp", dt);
		flattenHash.put("indexname", indexName);
		flattenHash.put("calcdatetime", calcDate);
//		String indexName = this._df2date.format(calcDate);
//		flattenHash.put("indexname", indexName);
		
		for (Map.Entry<String, Double> entry : ((HashMap<String, Double>)map.get("result")).entrySet()){
			flattenHash.put(entry.getKey(), entry.getValue());
		}
		
		collector.emit(new Values(flattenHash));
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("flatten"));
	}

}
